<!DOCTYPE html>
<html lang="zh-cn">
<head>
  
    <link type="text/css" rel="stylesheet" href="/bundles/blog-common.css?v=KOZafwuaDasEedEenI5aTy8aXH0epbm6VUJ0v3vsT_Q1"/>
<link id="MainCss" type="text/css" rel="stylesheet" href="/skins/ThinkInside/bundle-ThinkInside.css?v=RRjf6pEarGnbXZ86qxNycPfQivwSKWRa4heYLB15rVE1"/>
<link type="text/css" rel="stylesheet" href="/blog/customcss/428549.css?v=%2fam3bBTkW5NBWhBE%2fD0lcyJv5UM%3d"/>

</head>
<body>
<a name="top"></a>

<div id="page_begin_html"></div><script>load_page_begin_html();</script>

<div id="topics">
	<div class = "post">
		<h1 class = "postTitle">
			<a id="cb_post_title_url" class="postTitle2" href="https://www.cnblogs.com/frankdeng/p/9308585.html">Kafka集成SparkStreaming</a>
		</h1>
		<div class="clear"></div>
		<div class="postBody">
			<div id="cnblogs_post_body" class="blogpost-body"><h2 id="spark-streaming-kafka-integration-guide" class="title"><span>Spark Streaming + Kafka集成指南</span></h2>
<p><span><span>Kafka项目在版本0.8和0.10之间引入了一个新的消费者API，因此有两个独立的相应Spark Streaming包可用。</span><span>请选择正确的包，&nbsp;</span><span>请注意，0.8集成与后来的0.9和0.10代理兼容，但0.10集成与早期的代理不兼容。</span></span></p>
<p><strong>注意：从Spark 2.3.0开始，不推荐使用Kafka 0.8支持。</strong></p>
<p><span>Spark Streaming从Kafka接收数据，转换为spark streaming中的数据结构Dstream。数据接收方式有</span><span>两种 ：1 使用Receiver接收的旧方法：2使用Direct拉取的新方法（在Spark 1.3中引入）。</span></p>
<p><a href="https://spark.apache.org/docs/1.6.3/streaming-kafka-integration.html" target="_blank">https://spark.apache.org/docs/1.6.3/streaming-kafka-integration.html</a></p>
<p><a href="https://spark.apache.org/docs/2.3.1/streaming-kafka-0-10-integration.html" target="_blank"><span>https://spark.apache.org/docs/2.3.1/streaming-kafka-0-10-integration.html</span></a></p>
<h2 id="基于receiver的方式">Receiver方式</h2>
<p><span><span>&nbsp; &nbsp; &nbsp;Received是使用Kafka高级Consumer&nbsp;API实现的。</span><span>与所有接收器一样，</span></span>从Kafka通过Receiver接收的数据存储在Spark Executor的内存中，然后由Spark Streaming启动的job来处理数据。然而默认配置下，这种方式可能会因为底层的失败而丢失数据（请参阅<a href="https://spark.apache.org/docs/1.4.0/streaming-programming-guide.html#receiver-reliability">接收器可靠性</a>）。如果要启用高可靠机制，确保零数据丢失，要启用Spark Streaming的预写日志机制（Write Ahead Log，（已引入）在Spark 1.2）。该机制会同步地将接收到的Kafka数据保存到分布式文件系统（比如HDFS）上的预写日志中，以便底层节点在发生故障时也可以使用预写日志中的数据进行恢复。</p>
<p>如下图：</p>
<p><img src="https://images2018.cnblogs.com/blog/1385722/201807/1385722-20180715211535349-1558536234.png" alt="" /></p>
<p>接下来，我们将讨论如何在流应用程序中使用此方法。</p>
<h3><strong><span><span>1 链接&nbsp;</span></span></strong></h3>
<p><span><span>对于使用Maven项目定义的Scala / Java应用程序</span></span>时，我们需要添加相应的依赖包：</p>
<div class="cnblogs_code">
<pre>&lt;dependency&gt;&lt;!-- Spark Streaming Kafka --&gt;
    &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
    &lt;artifactId&gt;spark-streaming-kafka_2.<span style="color: #800080;">11</span>&lt;/artifactId&gt;
    &lt;version&gt;<span style="color: #800080;">1.6</span>.<span style="color: #800080;">3</span>&lt;/version&gt;
&lt;/dependency&gt;</pre>
</div>
<h3><strong><span><span>2 编程&nbsp;</span></span></strong></h3>
<p><span><span>在流应用程序代码中，导入</span></span><code>KafkaUtils</code><span><span>并创建输入DStream，如下所示。</span></span></p>
<p>Scala编程：</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">import org.apache.spark.streaming.kafka._

   val kafkaStream </span>=<span style="color: #000000;"> KafkaUtils.createStream(streamingContext, 
     [ZK quorum], [consumer group id], [per</span>-topic number of Kafka partitions to consume])</pre>
</div>
<p>Java编程</p>
<div class="cnblogs_code">
<pre> import org.apache.spark.streaming.kafka.*<span style="color: #000000;">;

 JavaPairReceiverInputDStream</span>&lt;String, String&gt; kafkaStream =<span style="color: #000000;"> 
     KafkaUtils.createStream(streamingContext,
     [ZK quorum], [consumer group id], [per</span>-topic number of Kafka partitions to consume]);</pre>
</div>
<p>还有几个需要注意的点：</p>
<ul>
<li>Kafka中topic的partition与Spark Streaming中生成的RDD的partition无关，因此，在KafkaUtils.createStream()中，增加某个topic的partition的数量，只会增加单个Receiver消费topic的线程数，也就是读取Kafka中topic partition的线程数量，它不会增加Spark在处理数据时的并行性。</li>
<li>可以使用不同的consumer group和topic创建多个Kafka输入DStream，以使用多个receiver并行接收数据。</li>
<li>如果已使用HDFS等复制文件系统启用了&ldquo;预读日志&rdquo;，则接收的数据已在日志中复制。因此，输入流的存储级别的存储级别<code>S<span style="font-family: 宋体;">torageLevel.MEMORY_AND_DISK_SER</span></code><span style="font-family: 宋体;">（即，使用<code>KafkaUtils.createStream(..., StorageLevel.MEMORY_AND_DISK_SER)</code>）。</span></li>
</ul>
<h3><span style="font-family: 宋体;"><strong><span><span>3 部署</span></span></strong></span></h3>
<p><span style="font-family: 宋体;"><span><span>与任何Spark应用程序一样，</span></span><code>spark-submit</code><span><span>用于启动应用程序。</span><span>但是，Scala / Java应用程序和Python应用程序的细节略有不同。</span></span></span></p>
<p><span>对于Scala和Java应用程序，如果您使用SBT或Maven进行项目管理，则将</span><code>spark-streaming-kafka_2.11</code><span><span>其及其依赖项</span><span>打包</span><span>到应用程序JAR中。</span><span>确保</span></span><code>spark-core_2.10</code><span>并</span><code>spark-streaming_2.10</code><span>标记为</span><code>provided</code><span><span>Spark安装中已存在的依赖项。</span><span>然后使用</span></span><code>spark-submit</code><span><span>启动应用程序</span></span><span>。</span></p>
<p><span>对于缺少SBT / Maven项目管理的Python应用程序，</span><code>spark-streaming-kafka_2.11</code><span>可以直接将其依赖项添加到</span><code>spark-submit</code><span>使用中</span><code>--packages</code><span><span>。</span><span>那是，</span></span></p>
<div class="cnblogs_code">
<pre> ./bin/spark-submit --packages org.apache.spark:spark-streaming-kafka_2.<span style="color: #800080;">11</span>:<span style="color: #800080;">1.6</span>.<span style="color: #800080;">3</span> ...</pre>
</div>
<p><span>另外，您也可以下载Maven构件的JAR&nbsp;</span><code>spark-streaming-kafka-assembly</code><span>从&nbsp;</span><a href="http://search.maven.org/#search|ga|1|a%3A%22spark-streaming-kafka-assembly_2.10%22%20AND%20v%3A%221.6.3%22">Maven仓库</a><span>，并将其添加到</span><code>spark-submit</code><span>用</span><code>--jars</code><span>。</span></p>
<h2 id="直接读取方式">Direct方式</h2>
<p>在spark1.3之后，引入了Direct方式。不同于Receiver的方式，Direct方式没有receiver这一层，其会周期性的获取Kafka中每个topic的每个partition中的最新offsets，之后根据设定的maxRatePerPartition来处理每个batch。其形式如下图：<br /><img src="https://images2015.cnblogs.com/blog/524764/201612/524764-20161228154450976-162798485.png" alt="" /><br />这种方法相较于Receiver方式的优势在于：</p>
<ul>
<li><strong>简化的并行</strong>：在Receiver的方式中我们提到创建多个Receiver之后利用union来合并成一个Dstream的方式提高数据传输并行度。而在Direct方式中，<strong>Kafka中的partition与RDD中的partition是一一对应</strong>的并行读取Kafka数据，这种映射关系也更利于理解和优化。</li>
<li><strong>高效</strong>：在Receiver的方式中，为了达到0数据丢失需要将数据存入Write Ahead Log中，这样在Kafka和日志中就保存了两份数据，浪费！而第二种方式不存在这个问题，只要我们Kafka的数据保留时间足够长，我们都能够从Kafka进行数据恢复。</li>
<li><strong>精确一次</strong>：在Receiver的方式中，使用的是Kafka的高阶API接口从Zookeeper中获取offset值，这也是传统的从Kafka中读取数据的方式，但由于Spark Streaming消费的数据和Zookeeper中记录的offset不同步，这种方式偶尔会造成数据重复消费。而第二种方式，直接使用了简单的低阶Kafka API，Offsets则利用Spark Streaming的checkpoints进行记录，消除了这种不一致性。</li>











</ul>
<p><span><span>请注意，此方法的一个缺点是它不会更新Zookeeper中的偏移量，因此基于Zookeeper的Kafka监视工具将不会显示进度。</span><span>但是，您可以在每个批处理中访问此方法处理的偏移量，并自行更新Zookeeper。</span></span></p>
<p><span>接下来，我们将讨论如何在流应用程序中使用此方法。</span></p>
<h3><span><strong><span><span>1 链接</span></span></strong></span></h3>
<div class="cnblogs_code">
<pre>&lt;dependency&gt;
    &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
    &lt;artifactId&gt;spark-streaming-kafka-<span style="color: #800080;">0</span>-10_2.<span style="color: #800080;">11</span>&lt;/artifactId&gt;
    &lt;version&gt;<span style="color: #800080;">2.3</span>.<span style="color: #800080;">1</span>&lt;/version&gt;
&lt;/dependency&gt;</pre>
</div>
<h3><span><strong><span><span><strong><span><span>2 编程</span></span></strong></span></span></strong></span></h3>
<p><span><strong><span><span><strong><span><span>请注意，导入的命名空间包括版本org.apache.spark.streaming.kafka010</span></span></strong></span></span></strong></span></p>
<p><span><strong><span><span><strong><span><span>Scala编程</span></span></strong></span></span></strong></span></p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

val kafkaParams </span>=<span style="color: #000000;"> Map[String, Object](
  </span><span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span> -&gt; <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">,
  </span><span style="color: #800000;">"</span><span style="color: #800000;">key.deserializer</span><span style="color: #800000;">"</span> -&gt;<span style="color: #000000;"> classOf[StringDeserializer],
  </span><span style="color: #800000;">"</span><span style="color: #800000;">value.deserializer</span><span style="color: #800000;">"</span> -&gt;<span style="color: #000000;"> classOf[StringDeserializer],
  </span><span style="color: #800000;">"</span><span style="color: #800000;">group.id</span><span style="color: #800000;">"</span> -&gt; <span style="color: #800000;">"</span><span style="color: #800000;">use_a_separate_group_id_for_each_stream</span><span style="color: #800000;">"</span><span style="color: #000000;">,
  </span><span style="color: #800000;">"</span><span style="color: #800000;">auto.offset.reset</span><span style="color: #800000;">"</span> -&gt; <span style="color: #800000;">"</span><span style="color: #800000;">latest</span><span style="color: #800000;">"</span><span style="color: #000000;">,
  </span><span style="color: #800000;">"</span><span style="color: #800000;">enable.auto.commit</span><span style="color: #800000;">"</span> -&gt; (<span style="color: #0000ff;">false</span><span style="color: #000000;">: java.lang.Boolean)
)

val topics </span>= Array(<span style="color: #800000;">"</span><span style="color: #800000;">topicA</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">topicB</span><span style="color: #800000;">"</span><span style="color: #000000;">)
val stream </span>=<span style="color: #000000;"> KafkaUtils.createDirectStream[String, String](
  streamingContext,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

stream.map(record </span>=&gt; (record.key, record.value))</pre>
</div>
<div class="codetabs tab-content">
<div id="tab_scala_0" class="tab-pane active" data-lang="scala">
<p><span>流中的每个项目都是</span><a href="http://kafka.apache.org/0100/javadoc/org/apache/kafka/clients/consumer/ConsumerRecord.html">ConsumerRecord</a>，有关可能的kafkaParams，请参阅<a href="http://kafka.apache.org/documentation.html#newconsumerconfigs">Kafka使用者配置文档</a>。如果Spark批处理持续时间大于默认的Kafka心跳会话超时（30秒），请适当增加heartbeat.interval.ms和session.timeout.ms。对于大于5分钟的批次，这将需要在代理上更改group.max.session.timeout.ms。请注意，该示例将enable.auto.commit设置为false，有关讨论，请参阅<a href="https://spark.apache.org/docs/2.3.1/streaming-kafka-0-10-integration.html#storing-offsets">存储偏移</a>。</p>
<h3>3&nbsp;Direct方式案例</h3>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.spark
 
import kafka.serializer.{StringDecoder, Decoder}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkContext, SparkConf}
 
import scala.reflect.ClassTag
 
</span><span style="color: #008000;">/*</span><span style="color: #008000;">*
  * Created by Administrator on 2018/7/28.
  </span><span style="color: #008000;">*/</span>
<span style="color: #0000ff;">object</span><span style="color: #000000;"> SparkStreamDemo {
  def main(args: Array[String]) {
 
    val conf </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> SparkConf()
    conf.setAppName(</span><span style="color: #800000;">"</span><span style="color: #800000;">spark_streaming</span><span style="color: #800000;">"</span><span style="color: #000000;">)
    conf.setMaster(</span><span style="color: #800000;">"</span><span style="color: #800000;">local[*]</span><span style="color: #800000;">"</span><span style="color: #000000;">)
 
    val sc </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> SparkContext(conf)
    sc.setCheckpointDir(</span><span style="color: #800000;">"</span><span style="color: #800000;">D:/checkpoints</span><span style="color: #800000;">"</span><span style="color: #000000;">)
    sc.setLogLevel(</span><span style="color: #800000;">"</span><span style="color: #800000;">ERROR</span><span style="color: #800000;">"</span><span style="color: #000000;">)
 
    val ssc </span>= <span style="color: #0000ff;">new</span> StreamingContext(sc, Seconds(<span style="color: #800080;">5</span><span style="color: #000000;">))
 
    </span><span style="color: #008000;">//</span><span style="color: #008000;"> val topics = Map("spark" -&gt; 2)</span>
<span style="color: #000000;"> 
    val kafkaParams </span>=<span style="color: #000000;"> Map[String, String](
      </span><span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span> -&gt; <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">,
      </span><span style="color: #800000;">"</span><span style="color: #800000;">group.id</span><span style="color: #800000;">"</span> -&gt; <span style="color: #800000;">"</span><span style="color: #800000;">spark</span><span style="color: #800000;">"</span><span style="color: #000000;">,
      </span><span style="color: #800000;">"</span><span style="color: #800000;">auto.offset.reset</span><span style="color: #800000;">"</span> -&gt; <span style="color: #800000;">"</span><span style="color: #800000;">smallest</span><span style="color: #800000;">"</span><span style="color: #000000;">
    )
    </span><span style="color: #008000;">//</span><span style="color: #008000;"> 直连方式拉取数据，这种方式不会修改数据的偏移量，需要手动的更新</span>
    val lines =  KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set(<span style="color: #800000;">"</span><span style="color: #800000;">spark</span><span style="color: #800000;">"</span><span style="color: #000000;">)).map(_._2)
    </span><span style="color: #008000;">//</span><span style="color: #008000;"> val lines = KafkaUtils.createStream(ssc, "node21:2181,node22:2181,node23:2181", "spark", topics).map(_._2)</span>
<span style="color: #000000;"> 
    val ds1 </span>= lines.flatMap(_.split(<span style="color: #800000;">"</span> <span style="color: #800000;">"</span>)).map((_, <span style="color: #800080;">1</span><span style="color: #000000;">))
 
    val ds2 </span>= ds1.updateStateByKey[Int]((x:Seq[Int], y:Option[Int]) =&gt;<span style="color: #000000;"> {
      Some(x.sum </span>+ y.getOrElse(<span style="color: #800080;">0</span><span style="color: #000000;">))
    })
 
    ds2.print()
 
    ssc.start()
    ssc.awaitTermination()
 
  }
}</span></pre>
</div>
</div>
</div>
<h2 id="spark向kafka中写入数据">Spark向kafka中写入数据</h2>
<p>上文阐述了Spark如何从Kafka中流式的读取数据，下面我整理向Kafka中写数据。与读数据不同，Spark并没有提供统一的接口用于写入Kafka，所以我们需要使用底层Kafka接口进行包装。<br />最直接的做法我们可以想到如下这种方式：</p>
<div class="cnblogs_code">
<pre>input.foreachRDD(rdd =&gt;
  <span style="color: #008000;">//</span><span style="color: #008000;"> 不能在这里创建KafkaProducer</span>
  rdd.foreachPartition(partition =&gt;<span style="color: #000000;">
    partition.</span><span style="color: #0000ff;">foreach</span><span style="color: #000000;">{
      </span><span style="color: #0000ff;">case</span> x:String=&gt;<span style="color: #000000;">{
        val props </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> HashMap[String, Object]()
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          </span><span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">)
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
          </span><span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">)
        println(x)
        val producer </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> KafkaProducer[String,String](props)
        val message</span>=<span style="color: #0000ff;">new</span> ProducerRecord[String, String](<span style="color: #800000;">"</span><span style="color: #800000;">output</span><span style="color: #800000;">"</span>,<span style="color: #0000ff;">null</span><span style="color: #000000;">,x)
        producer.send(message)
      }
    }
  )
) </span></pre>
</div>
<p>但是这种方式缺点很明显，对于每个partition的每条记录，我们都需要创建KafkaProducer，然后利用producer进行输出操作，注意这里我们并不能将KafkaProducer的新建任务放在foreachPartition外边，因为KafkaProducer是不可序列化的（not serializable）。显然这种做法是不灵活且低效的，因为每条记录都需要建立一次连接。如何解决呢？</p>
<p>1.首先，我们需要将KafkaProducer利用lazy val的方式进行包装如下：</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">import java.util.concurrent.Future
import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerRecord, RecordMetadata }
</span><span style="color: #0000ff;">class</span> KafkaSink[K, V](createProducer: () =&gt;<span style="color: #000000;"> KafkaProducer[K, V]) extends Serializable {
  </span><span style="color: #008000;">/*</span><span style="color: #008000;"> This is the key idea that allows us to work around running into
     NotSerializableExceptions. </span><span style="color: #008000;">*/</span><span style="color: #000000;">
  lazy val producer </span>=<span style="color: #000000;"> createProducer()
  def send(topic: String, key: K, value: V): Future[RecordMetadata] </span>=<span style="color: #000000;">
    producer.send(</span><span style="color: #0000ff;">new</span><span style="color: #000000;"> ProducerRecord[K, V](topic, key, value))
  def send(topic: String, value: V): Future[RecordMetadata] </span>=<span style="color: #000000;">
    producer.send(</span><span style="color: #0000ff;">new</span><span style="color: #000000;"> ProducerRecord[K, V](topic, value))
}

</span><span style="color: #0000ff;">object</span><span style="color: #000000;"> KafkaSink {
  import scala.collection.JavaConversions._
  def apply[K, V](config: Map[String, Object]): KafkaSink[K, V] </span>=<span style="color: #000000;"> {
    val createProducerFunc </span>= () =&gt;<span style="color: #000000;"> {
      val producer </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> KafkaProducer[K, V](config)
      sys.addShutdownHook {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> Ensure that, on executor JVM shutdown, the Kafka producer sends
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> any buffered messages to Kafka before shutting down.</span>
<span style="color: #000000;">        producer.close()
      }
      producer
    }
    </span><span style="color: #0000ff;">new</span><span style="color: #000000;"> KafkaSink(createProducerFunc)
  }
  def apply[K, V](config: java.util.Properties): KafkaSink[K, V] </span>=<span style="color: #000000;"> apply(config.toMap)
}</span></pre>
</div>
<p>2.之后我们利用广播变量的形式，将KafkaProducer广播到每一个executor，如下：</p>
<div class="cnblogs_code">
<pre><span style="color: #008000;">//</span><span style="color: #008000;"> 广播KafkaSink</span>
val kafkaProducer: Broadcast[KafkaSink[String, String]] =<span style="color: #000000;"> {
  val kafkaProducerConfig </span>=<span style="color: #000000;"> {
    val p </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties()
    p.setProperty(</span><span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span><span style="color: #000000;">, Conf.brokers)
    p.setProperty(</span><span style="color: #800000;">"</span><span style="color: #800000;">key.serializer</span><span style="color: #800000;">"</span><span style="color: #000000;">, classOf[StringSerializer].getName)
    p.setProperty(</span><span style="color: #800000;">"</span><span style="color: #800000;">value.serializer</span><span style="color: #800000;">"</span><span style="color: #000000;">, classOf[StringSerializer].getName)
    p
  }
  log.warn(</span><span style="color: #800000;">"</span><span style="color: #800000;">kafka producer init done!</span><span style="color: #800000;">"</span><span style="color: #000000;">)
  ssc.sparkContext.broadcast(KafkaSink[String, String](kafkaProducerConfig))
}</span></pre>
</div>
<p>这样我们就能在每个executor中愉快的将数据输入到kafka当中：</p>
<div class="cnblogs_code">
<pre><span style="color: #008000;">//</span><span style="color: #008000;">输出到kafka</span>
segmentedStream.foreachRDD(rdd =&gt;<span style="color: #000000;"> {
  </span><span style="color: #0000ff;">if</span> (!<span style="color: #000000;">rdd.isEmpty) {
    rdd.</span><span style="color: #0000ff;">foreach</span>(record =&gt;<span style="color: #000000;"> {
      kafkaProducer.value.send(Conf.outTopics, record._1.toString, record._2)
      </span><span style="color: #008000;">//</span><span style="color: #008000;"> do something else</span>
<span style="color: #000000;">    })
  }
})</span></pre>
</div>
<h2 id="spark-streamingkafka应用">Spark streaming+Kafka应用</h2>
<p>一般Spark Streaming进行流式处理，首先利用上文我们阐述的Direct方式从Kafka拉取batch，之后经过分词、统计等相关处理，回写到DB上（一般为Hbase或者Mysql），由此高效实时的完成每天大量数据的词频统计任务。</p>
<h2 id="spark-streamingkafka调优">Spark streaming+Kafka调优</h2>
<p>Spark streaming+Kafka的使用中，当数据量较小，很多时候默认配置和使用便能够满足情况，但是当数据量大的时候，就需要进行一定的调整和优化，而这种调整和优化本身也是不同的场景需要不同的配置。</p>
<h3 id="合理的批处理时间batchduration">1 合理的批处理时间（batchDuration）</h3>
<p>几乎所有的Spark Streaming调优文档都会提及批处理时间的调整，在StreamingContext初始化的时候，有一个参数便是批处理时间的设定。如果这个值设置的过短，即个batchDuration所产生的Job并不能在这期间完成处理，那么就会造成数据不断堆积，最终导致Spark Streaming发生阻塞。而且，一般对于batchDuration的设置不会小于500ms，因为过小会导致SparkStreaming频繁的提交作业，对整个streaming造成额外的负担。在平时的应用中，根据不同的应用场景和硬件配置，我设在1~10s之间，我们可以根据SparkStreaming的可视化监控界面，观察Total Delay来进行batchDuration的调整，如下图：<br /><img src="https://images2015.cnblogs.com/blog/524764/201701/524764-20170103171142066-1054088549.png" alt="" /></p>
<h3 id="合理的kafka拉取量maxrateperpartition重要">2 合理的Kafka拉取量（maxRatePerPartition重要）</h3>
<p>对于Spark Streaming消费kafka中数据的应用场景，这个配置是非常关键的，配置参数为：spark.streaming.kafka.maxRatePerPartition。这个参数默认是没有上线的，即kafka当中有多少数据它就会直接全部拉出。而根据生产者写入Kafka的速率以及消费者本身处理数据的速度，同时这个参数需要结合上面的batchDuration，使得每个partition拉取在每个batchDuration期间拉取的数据能够顺利的处理完毕，做到尽可能高的吞吐量，而这个参数的调整可以参考可视化监控界面中的Input Rate和Processing Time，如下图：<br /><img src="https://images2015.cnblogs.com/blog/524764/201701/524764-20170103172311159-1621531817.png" alt="" /><br /><img src="https://images2015.cnblogs.com/blog/524764/201701/524764-20170103172338941-153218465.png" alt="" /></p>
<h3 id="缓存反复使用的dstreamrdd">3 缓存反复使用的Dstream（RDD）</h3>
<p>Spark中的RDD和SparkStreaming中的Dstream，如果被反复的使用，最好利用cache()，将该数据流缓存起来，防止过度的调度资源造成的网络开销。可以参考观察Scheduling Delay参数，如下图：<br /><img src="https://images2015.cnblogs.com/blog/524764/201701/524764-20170103185139331-147812384.png" alt="" /></p>
<h3 id="设置合理的gc">4 设置合理的GC</h3>
<p>长期使用Java的小伙伴都知道，JVM中的垃圾回收机制，可以让我们不过多的关注与内存的分配回收，更加专注于业务逻辑，JVM都会为我们搞定。对JVM有些了解的小伙伴应该知道，在Java虚拟机中，将内存分为了初生代（eden generation）、年轻代（young generation）、老年代（old generation）以及永久代（permanent generation），其中每次GC都是需要耗费一定时间的，尤其是老年代的GC回收，需要对内存碎片进行整理，通常采用标记-清楚的做法。同样的在Spark程序中，JVM GC的频率和时间也是影响整个Spark效率的关键因素。在通常的使用中建议：</p>
<pre><code class="hljs cpp">--conf <span class="hljs-string">"spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC"</span></code></pre>
<h3 id="设置合理的cpu资源数">5 设置合理的CPU资源数</h3>
<p>CPU的core数量，每个executor可以占用一个或多个core，可以通过观察CPU的使用率变化来了解计算资源的使用情况，例如，很常见的一种浪费是一个executor占用了多个core，但是总的CPU使用率却不高（因为一个executor并不总能充分利用多核的能力），这个时候可以考虑让么个executor占用更少的core，同时worker下面增加更多的executor，或者一台host上面增加更多的worker来增加并行执行的executor的数量，从而增加CPU利用率。但是增加executor的时候需要考虑好内存消耗，因为一台机器的内存分配给越多的executor，每个executor的内存就越小，以致出现过多的数据spill over甚至out of memory的情况。</p>
<h3 id="设置合理的parallelism">6 设置合理的parallelism</h3>
<p>partition和parallelism，partition指的就是数据分片的数量，每一次task只能处理一个partition的数据，这个值太小了会导致每片数据量太大，导致内存压力，或者诸多executor的计算能力无法利用充分；但是如果太大了则会导致分片太多，执行效率降低。在执行action类型操作的时候（比如各种reduce操作），partition的数量会选择parent RDD中最大的那一个。而parallelism则指的是在RDD进行reduce类操作的时候，默认返回数据的paritition数量（而在进行map类操作的时候，partition数量通常取自parent RDD中较大的一个，而且也不会涉及shuffle，因此这个parallelism的参数没有影响）。所以说，这两个概念密切相关，都是涉及到数据分片的，作用方式其实是统一的。通过spark.default.parallelism可以设置默认的分片数量，而很多RDD的操作都可以指定一个partition参数来显式控制具体的分片数量。<br />在SparkStreaming+kafka的使用中，我们采用了Direct连接方式，前文阐述过Spark中的partition和Kafka中的Partition是一一对应的，我们一般默认设置为Kafka中Partition的数量。</p>
<h3 id="使用高性能的算子">7 使用高性能的算子</h3>
<p>这里参考了美团技术团队的博文，并没有做过具体的性能测试，其建议如下：</p>
<ul>
<li>使用reduceByKey/aggregateByKey替代groupByKey</li>
<li>使用mapPartitions替代普通map</li>
<li>使用foreachPartitions替代foreach</li>
<li>使用filter之后进行coalesce操作</li>
<li>使用repartitionAndSortWithinPartitions替代repartition与sort类操作</li>











</ul>
<h3 id="使用kryo优化序列化性能">8 使用Kryo优化序列化性能</h3>
<p>这个优化原则我本身也没有经过测试，但是好多优化文档有提到，这里也记录下来。<br />在Spark中，主要有三个地方涉及到了序列化：</p>
<ul>
<li>在算子函数中使用到外部变量时，该变量会被序列化后进行网络传输（见&ldquo;原则七：广播大变量&rdquo;中的讲解）。</li>
<li>将自定义的类型作为RDD的泛型类型时（比如JavaRDD，Student是自定义类型），所有自定义类型对象，都会进行序列化。因此这种情况下，也要求自定义的类必须实现Serializable接口。</li>
<li>使用可序列化的持久化策略时（比如MEMORY_ONLY_SER），Spark会将RDD中的每个partition都序列化成一个大的字节数组。</li>











</ul>
<p>对于这三种出现序列化的地方，我们都可以通过使用Kryo序列化类库，来优化序列化和反序列化的性能。Spark默认使用的是Java的序列化机制，也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。但是Spark同时支持使用Kryo序列化库，Kryo序列化类库的性能比Java序列化类库的性能要高很多。官方介绍，Kryo序列化机制比Java序列化机制，性能高10倍左右。Spark之所以默认没有使用Kryo作为序列化类库，是因为Kryo要求最好要注册所有需要进行序列化的自定义类型，因此对于开发者来说，这种方式比较麻烦。</p>
<p>以下是使用Kryo的代码示例，我们只要设置序列化类，再注册要序列化的自定义类型即可（比如算子函数中使用到的外部变量类型、作为RDD泛型类型的自定义类型等）：</p>
<div class="cnblogs_code">
<pre><span style="color: #008000;">//</span><span style="color: #008000;"> 创建SparkConf对象。</span>
val conf = <span style="color: #0000ff;">new</span><span style="color: #000000;"> SparkConf().setMaster(...).setAppName(...)
</span><span style="color: #008000;">//</span><span style="color: #008000;"> 设置序列化器为KryoSerializer。</span>
conf.<span style="color: #0000ff;">set</span>(<span style="color: #800000;">"</span><span style="color: #800000;">spark.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.spark.serializer.KryoSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">)
</span><span style="color: #008000;">//</span><span style="color: #008000;"> 注册要序列化的自定义类型。</span>
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))</pre>
</div>
<h3 id="结果">结果</h3>
<p>经过种种调试优化，我们最终要达到的目的是，Spark Streaming能够实时的拉取Kafka当中的数据，并且能够保持稳定，如下图所示：<br /><img src="https://images2015.cnblogs.com/blog/524764/201701/524764-20170103174522644-1161173395.png" alt="" /></p>
<p>当然不同的应用场景会有不同的图形，这是本文词频统计优化稳定后的监控图，我们可以看到Processing Time这一柱形图中有一Stable的虚线，而大多数Batch都能够在这一虚线下处理完毕，说明整体Spark Streaming是运行稳定的。</p></div><div id="MySignature"></div>


</body>
</html>
